package com.chenwc.rocketmq.example;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


/**
 * 异步发送消息
 * 异步消息发送指的是生产者发送消息后，不需要等待Broker（消息队列服务器）的响应即可继续执行后续操作，而消息的发送结果会通过回调函数进行通知。
 * 这种方式提高了系统的吞吐量和响应速度，特别适用于高并发和实时性要求较高的场景。
 *
 * @author chenwc
 */
public class AsyncProducerExample {
    private static final Logger logger = LoggerFactory.getLogger(AsyncProducerExample.class);

    private AsyncProducerExample() {
    }

    public static void main(String[] args) throws InterruptedException, MQClientException {
        String topic = "TopicTest";
        final DefaultMQProducer producer = ProducerSingleton.getInstance();

        // 定义消息体
        byte[] body = "This is a async message for Apache RocketMQ".getBytes(StandardCharsets.UTF_8);
        // 定义消息标签
        String tag = "TagA";
        // 定义消息key
        String keys = UUID.randomUUID().toString().toUpperCase().replace("-", "");
        // 创建消息对象
        final Message message = new Message(topic, tag, keys, body);

        // 设置异步发送的线程池
        ExecutorService sendCallbackExecutor = Executors.newCachedThreadPool();
        producer.setAsyncSenderExecutor(sendCallbackExecutor);
        try {
            // 异步发送消息1
            producer.send(message, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    logger.info("Send message successfully, {}", sendResult.toString());
                }

                @Override
                public void onException(Throwable e) {
                    logger.error("Failed to send message", e);
                }
            });

            // 异步发送消息2
            producer.send(message, new SendCallbackImpl());

        } catch (Throwable t) {
            logger.error("Failed to send message", t);
        }
        logger.info("start asynchronous send");

        // 阻止以避免后台线程的存在
        Thread.sleep(Long.MAX_VALUE);
        // 关闭生产者
        ProducerSingleton.closeInstance();
    }

    /**
     * 实现SendCallback接口
     */
    static class SendCallbackImpl implements SendCallback {

        @Override
        public void onSuccess(SendResult sendResult) {
            logger.info("Send message successfully, {}", sendResult.toString());
        }

        @Override
        public void onException(Throwable e) {
            logger.error("Failed to send message", e);
        }
    }
}
